#pragma once

#include <memory>

#include "common/hash_util.h"
#include "common/node_utils.h"
#include "nodes/execnodes.h"
#include "nodes/primnodes.h"
#include "planner/operator.h"
extern "C" {
#include <nodes/nodes.h>
#include <nodes/parsenodes.h>
#include <nodes/pg_list.h>
}

struct RangeTblEntry;

namespace optimizer {

class OperatorVisitor;

/**
 * Operator that represents another group
 */
class LeafOperator : public Operator {
 public:
  static constexpr const OperatorType TYPE = OperatorType::LeafOperator;
  explicit LeafOperator(group_id_t group) : Operator(OperatorType::LeafOperator), origin_group_(group) {}

  ~LeafOperator() override = default;

  std::shared_ptr<Operator> Copy() const override { return std::make_shared<LeafOperator>(origin_group_); }

  bool operator==(const Operator &r) const override {
    if (Operator::operator==(r)) {
      const auto &other = CastTo<LeafOperator>();
      return origin_group_ == other.origin_group_;
    }
    return false;
  }

  hash_t Hash() const override { return HashUtil::CombineHashes(Operator::Hash(), HashUtil::Hash(origin_group_)); }

  group_id_t GetOriginGroup() const { return origin_group_; }

  std::string ToString() const override { return std::format("(leaf: {})", origin_group_); }

  void Accept(OperatorVisitor *v) override;

 private:
  group_id_t origin_group_;
};

class LogicalGet : public Operator {
 public:
  static constexpr const OperatorType TYPE = OperatorType::LogicalGet;

  explicit LogicalGet(RangeTblEntry *rte) : Operator(OperatorType::LogicalGet), rte_(rte) {}

  ~LogicalGet() override = default;

  std::shared_ptr<Operator> Copy() const override { return std::make_shared<LogicalGet>(*this); }

  bool operator==(const Operator &r) const override {
    if (Operator::operator==(r)) {
      const auto &get = CastTo<LogicalGet>();
      return equal(rte_, get.rte_);
    }
    return false;
  }

  hash_t Hash() const override { return HashUtil::CombineHashes(Operator::Hash(), HashUtil::Hash(GetTableName())); }

  std::string ToString() const override { return std::format("(scan: {})", GetTableName()); }

  std::string GetTableName() const {
    if (rte_ != nullptr) {
      if (rte_->alias != nullptr)
        return rte_->alias->aliasname;
      if (rte_->eref != nullptr)
        return rte_->eref->aliasname;
    }
    return "dummy_table";
  }

  void Accept(OperatorVisitor *v) override;

  RangeTblEntry *GetRangeTblEntry() const { return rte_; }

 private:
  RangeTblEntry *rte_{nullptr};
};

/**
 * Logical operator for LogicalLimit
 * This supports embedded ORDER BY information
 */
class LogicalLimit : public Operator {
 public:
  static constexpr const OperatorType TYPE = OperatorType::LogicalLimit;

  LogicalLimit(Expression *limit, Expression *offset) : Operator(OperatorType::LogicalLimit) {
    AddExpression(limit);
    AddExpression(offset);
  }

  std::string ToString() const override {
    auto *limit = linitial(expressions_);
    if (list_length(expressions_) == 2) {
      auto *offset = lsecond(expressions_);
      return std::format("(limit (limit: {} offset: {}))", NodeUtils::ToString(limit), NodeUtils::ToString(offset));
    }
    return std::format("(limit (limit: {}))", NodeUtils::ToString(limit));
  }

  void Accept(OperatorVisitor *v) override;
};

class LogicalSetOperation : public Operator {
 public:
  static constexpr const OperatorType TYPE = OperatorType::UNDEFINED;

  LogicalSetOperation(OperatorType type, bool is_all) : Operator(type), is_all_(is_all) {}

  std::shared_ptr<Operator> Copy() const override;
  hash_t Hash() const override;
  bool operator==(const Operator &r) const override;

  std::string ToString() const override { return std::format("(setop: )"); }
  void Accept(OperatorVisitor *v) override;

 private:
  bool is_all_{false};
};

class LogicalJoin : public Operator {
 public:
  constexpr static const OperatorType TYPE = OperatorType::LogicalJoin;

  explicit LogicalJoin(JoinType join_type, List *condition)
      : Operator(OperatorType::LogicalJoin), join_type_(join_type) {
    SetFilter(condition);
  }

  std::shared_ptr<Operator> Copy() const override;
  hash_t Hash() const override;
  bool operator==(const Operator &r) const override;

  std::string ToString() const override {
    std::string join_type_str;
    switch (join_type_) {
      case JOIN_INNER:
        join_type_str = "inner";
        break;
      case JOIN_LEFT:
        join_type_str = "left";
        break;
      case JOIN_RIGHT:
        join_type_str = "right";
        break;
      case JOIN_FULL:
        join_type_str = "full";
        break;
      case JOIN_SEMI:
        join_type_str = "semi";
        break;
      case JOIN_ANTI:
        join_type_str = "anti";
        break;
      case JOIN_RIGHT_ANTI:
        join_type_str = "right_anti";
        break;
      case JOIN_UNIQUE_OUTER:
        join_type_str = "unique_outer";
        break;
      case JOIN_UNIQUE_INNER:
        join_type_str = "unique_inner";
        break;
      default:
        join_type_str = "unknown";
        break;
    }

    auto str = std::format("(join ({} {}))", join_type_str, NodeUtils::ToString(expressions_));

    return str;
  }
  void Accept(OperatorVisitor *v) override;

 private:
  JoinType join_type_;
};

class LogicalAggregate : public Operator {
 public:
  enum class AggExprState { ONLY_AGG_EXPR, ONLY_GROUP_EXPR, AGG_AND_GROUP_EXPR };
  constexpr static const OperatorType TYPE = OperatorType::LogicalAggregate;

  LogicalAggregate(List *group_expr, List *agg_expr) : Operator(OperatorType::LogicalAggregate) {
    if (group_expr != nullptr)
      AddExpression((Expr *)group_expr);
    if (agg_expr != nullptr)
      AddExpression((Expr *)agg_expr);

    if ((group_expr != nullptr) && (agg_expr != nullptr))
      agg_expr_state_ = AggExprState::AGG_AND_GROUP_EXPR;
    else if (group_expr != nullptr)
      agg_expr_state_ = AggExprState::ONLY_GROUP_EXPR;
    else
      agg_expr_state_ = AggExprState::ONLY_AGG_EXPR;
  }

  std::shared_ptr<Operator> Copy() const override;
  hash_t Hash() const override;
  bool operator==(const Operator &r) const override;

  List *GetGroupExpressions() const {
    if (agg_expr_state_ != AggExprState::ONLY_AGG_EXPR)
      return linitial_node(List, expressions_);
    return nullptr;
  }

  List *GetAggExpressions() const {
    if (agg_expr_state_ == AggExprState::AGG_AND_GROUP_EXPR)
      return lsecond_node(List, expressions_);
    if (agg_expr_state_ == AggExprState::ONLY_AGG_EXPR)
      return linitial_node(List, expressions_);
    return nullptr;
  }

  std::string ToString() const override {
    return std::format("(agg: group: {} agg: {})", NodeUtils::ToString(GetGroupExpressions()),
                       NodeUtils::ToString(GetAggExpressions()));
  }
  void Accept(OperatorVisitor *v) override;

 private:
  AggExprState agg_expr_state_;
};

}  // namespace optimizer
